package com.jinken._04delay;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 消费者
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        //创建消费者对象
        DefaultMQPushConsumer dmc = new DefaultMQPushConsumer("delay_consumer_group1");
        //设置服务地址
        dmc.setNamesrvAddr("192.168.126.100:9876");
        //设置订阅
        dmc.subscribe("delay_topic1","*");

        //监听消息
        dmc.setMessageListener(new MessageListenerConcurrently() {
            /**
             *
             * @param list 消息列表
             * @param consumeConcurrentlyContext 消费者上下文对象
             * @return 返回消费状态，
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                for (MessageExt messageExt : list) {
                    String msg = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                    System.out.println("msg = " + msg);
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //启动消费者
        dmc.start();
        //永久执行
        Thread.sleep(Long.MAX_VALUE);
    }
}
